package com.wuwangfu.async;


import com.wuwangfu.func.AsyncMySQLFunc;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-06-29
 */
public class AsyncQueryFromMySQL {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置job重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000));
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        int capacity = 50;
        AsyncDataStream.orderedWait(
                lines,//输入的数据流
                new AsyncMySQLFunc(capacity),//异步查询的Function实例
                3000,//超时时间
                TimeUnit.MILLISECONDS,//时间单位
                capacity//异步请求最大数量，不传该参数默认值为100
        ).print();

        env.execute();
    }
}
